package org.fujene.memoryblock.optimizer;

import java.util.HashSet;
import java.util.Hashtable;
import java.util.Map.Entry;
import org.fujene.communicate.deleter.DeleterPackage;
import org.fujene.constants.Constants;
import org.fujene.memcon.DiskController;
import org.fujene.memcon.MemoryAllocator;
import org.fujene.memcon.MemoryController;
import org.fujene.memman.DeleteFilter;
import org.fujene.memoryblock.content.ContentBlock;
import org.fujene.memoryblock.content.OptimizerContentBlock;
import org.fujene.memoryblock.invert.OptimizerInvertIndexBlock;
import org.fujene.structure.ContentInfo;
import org.fujene.structure.LookupPointer;

public class DeleteOptimizer extends Optimizer{
	private DeleteFilter filter;
	private MemoryController controller;
	private MemoryAllocator allocator;
	private DiskController disk;
	private ContentInfo contentInfo;

	public DeleteOptimizer(DeleteFilter filter, MemoryController controller, MemoryAllocator allocator,
		DiskController diskControl, ContentInfo contentInfo){
		this.filter = filter;
		this.controller = controller;
		this.allocator = allocator;
		this.disk = diskControl;
		this.contentInfo = contentInfo;
	}

	private HashSet<Long> obtainContentIDByFlag(int cluster){
		LookupPointer currentPoint = allocator.getFirstPointer(cluster, Constants.MEMORY_TYPE_INVINDEX, 0);
		OptimizerInvertIndexBlock invBlock =
			(OptimizerInvertIndexBlock)allocator.dispense(cluster, currentPoint.block());
		HashSet<Long> mergedIDs = new HashSet<Long>();

		while(invBlock != null){
			HashSet<Long> pickedIDs = invBlock.pickIDbyFlag(filter);
			if(pickedIDs != null) mergedIDs.addAll(pickedIDs);

			invBlock = (OptimizerInvertIndexBlock)allocator.getNextBlock(invBlock);
		}

		return mergedIDs;
	}

	private Hashtable<LookupPointer, Integer> pickContentByIDFlag(HashSet<Long> localFilter, int cluster,
		int subtype){
		Hashtable<LookupPointer, Integer> pointList = new Hashtable<LookupPointer, Integer>();
		LookupPointer curContentPoint =
			allocator.getFirstPointer(cluster, Constants.MEMORY_TYPE_CONTENTS, subtype);

		OptimizerContentBlock block =
			new OptimizerContentBlock(disk.getBuffer(cluster, curContentPoint.block()));
		while(curContentPoint.point() != -1){
			// filter with every ID in the block
			int length = block.checkByID(localFilter, curContentPoint.point());
			if(length != -1) pointList.put(curContentPoint.withCluster(cluster), Integer.valueOf(length));
			curContentPoint.modify(curContentPoint.point(), block.getNextOne(curContentPoint.point()));

			if(curContentPoint.point() == -1){
				if(block.getOption(ContentBlock.OPT_NEXT) == -1) break;
				else{
					curContentPoint.modify(block.getOption(ContentBlock.OPT_NEXT), 0);
					block = new OptimizerContentBlock(disk.getBuffer(cluster, curContentPoint.block()));
				}
			}
		}

		return pointList;
	}

	private Hashtable<LookupPointer, Integer> pickContentByID(int cluster, int subtype){
		Hashtable<LookupPointer, Integer> pointList = new Hashtable<LookupPointer, Integer>();
		LookupPointer currentPoint =
			allocator.getFirstPointer(cluster, Constants.MEMORY_TYPE_CONTENTS, subtype);

		OptimizerContentBlock block =
			new OptimizerContentBlock(disk.getBuffer(cluster, currentPoint.block()));
		while(currentPoint.point() != -1){
			// filter with every ID in the block
			int length = block.checkByID(filter, currentPoint.point());
			if(length != -1) pointList.put(currentPoint.withCluster(cluster), Integer.valueOf(length));
			currentPoint.modify(currentPoint.point(), block.getNextOne(currentPoint.point()));

			if(currentPoint.point() == -1){
				if(block.getOption(ContentBlock.OPT_NEXT) == -1) break;
				else{
					currentPoint.modify(block.getOption(ContentBlock.OPT_NEXT), 0);
					block = new OptimizerContentBlock(disk.getBuffer(cluster, currentPoint.block()));
				}
			}
		}

		return pointList;
	}

	private Hashtable<LookupPointer, Integer> pickContentByString(int cluster, int subtype){
		Hashtable<LookupPointer, Integer> pointList = new Hashtable<LookupPointer, Integer>();
		LookupPointer currentPoint =
			allocator.getFirstPointer(cluster, Constants.MEMORY_TYPE_CONTENTS, subtype);

		OptimizerContentBlock block =
			new OptimizerContentBlock(disk.getBuffer(cluster, currentPoint.block()));
		while(currentPoint.point() != -1){
			// filter with every ID in the block
			int length = block.checkByContent(filter, currentPoint.point());
			if(length != -1) pointList.put(currentPoint.withCluster(cluster), Integer.valueOf(length));
			currentPoint.modify(currentPoint.point(), block.getNextOne(currentPoint.point()));

			if(currentPoint.point() == -1){
				if(block.getOption(ContentBlock.OPT_NEXT) == -1) break;
				else{
					currentPoint.modify(block.getOption(ContentBlock.OPT_NEXT), 0);
					block = new OptimizerContentBlock(disk.getBuffer(cluster, currentPoint.block()));
				}
			}
		}

		return pointList;
	}

	private Hashtable<LookupPointer, Integer> pickContentByDeleteKey(int cluster, int subtype){
		switch(filter.deleteKeyType()){
		case DeleterPackage.TYPE_ID:
			return pickContentByID(cluster, subtype);
		case DeleterPackage.TYPE_CONTENT:
			return pickContentByString(cluster, subtype);
		case DeleterPackage.TYPE_FLAG:
			HashSet<Long> filterIDs = obtainContentIDByFlag(cluster);
			return pickContentByIDFlag(filterIDs, cluster, subtype);
		default:
		}
		return null;
	}

	private void deleteContent(Hashtable<LookupPointer, Integer> pointList, int cluster, int subtype){
		OptimizerContentBlock contentBlock, nextContentBlock;
		int buffer = 0;
		LookupPointer prev, next;
		prev = allocator.getFirstPointer(cluster, Constants.MEMORY_TYPE_CONTENTS, subtype);
		contentBlock = new OptimizerContentBlock(disk.getBuffer(cluster, prev.block()));
		for(Entry<LookupPointer, Integer> point: pointList.entrySet()){
			next = point.getKey();
			if(prev.block() != next.block()){
				// Move from another block
				nextContentBlock = new OptimizerContentBlock(disk.getBuffer(cluster, next.block()));
				int neededSize = nextContentBlock.getAvailSize(buffer);
				byte[] val = nextContentBlock.getRawBody(0, neededSize);
				contentBlock.setRawBody(prev.point(), val);
				buffer = neededSize;
				contentBlock = nextContentBlock;
			}else{
				contentBlock.moveBody(prev.block(), next.block(), buffer);
				buffer += point.getValue().intValue();
			}
			prev = next;
		}
	}

	@Override
	public void optimize(){
		// Stop indexing, suspend node
		controller.suspendMyNode();

		// Add block information to each block
		allocator.saveAllMemory();

		for(int cluster = allocator.clusterID(0), i = 0; cluster != -1; i++, cluster = allocator.clusterID(i)){
			for(int j = 0; j < contentInfo.contentsNum(); j++){
				Hashtable<LookupPointer, Integer> pointList = pickContentByDeleteKey(cluster, j);
				deleteContent(pointList, cluster, j);
			}
		}

		controller.resumeMyNode();
	}
}
